
#include <iostream>
#include <cstdio>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/syscall.h> /*必须引用这个文件 */
#include <regex>

#include "../../pemd/include/app.h"
#include "../include/DataCenterManager.h"
#include "../../db/include/CDBManager.h"
#include "../../../include/DCData.h"
#include "../../../include/protocol_engine_config.h"

#include "../../modbus/include/modbus_datastruct.h"
#include "../../../src/logger/include/CLogger.h"
#include "../../common/include/common.h"

#include "../include/CangZhouRptPatch.h"

//模型设置主题
#define  DATACENTER_MODEL_SET_PUB  "pemd/set/request/database/model"
#define  DATACENTER_MODEL_SET_SUB  "database/set/response/pemd/model"
//模型删除主题
#define  DATACENTER_MODEL_DELE_PUB "pemd/action/request/database/deletemodel"
#define  DATACENTER_MODEL_DELE_SUB "database/action/response/pemd/deletemodel"
//模型查询主题
#define  DATACENTER_MODEL_QUERY_PUB "pemd/get/request/database/model"
#define  DATACENTER_MODEL_QUERY_SUB  "database/get/response/pemd/model"
//模型名称查询主题
#define  DATACENTER_MODEL_NAME_QUERY_PUB  "pemd/get/request/database/modelschema"
#define  DATACENTER_MODEL_NAME_QUERY_SUB  "database/get/response/pemd/modelschema"

//设备注册
#define  DATACENTER_DEV_REGISTER_PUB "pemd/set/request/database/register"
#define  DATACENTER_DEV_REGISTER_SUB "database/set/response/pemd/register"
//设备注册通知
#define  DATACENTER_DEV_REGISTER_NOTIFY_SUB  "database/notify/spont/allApp/Updateregister"
//注册关系查询
#define  DATACENTER_DEV_REGISTER_RELATION_PUB "pemd/get/request/database/register"
#define  DATACENTER_DEV_REGISTER_RELATION_SUB "database/get/response/pemd/register"

//设备取消注册
#define  DATACENTER_DEV_UNREGISTER_PUB "pemd/action/request/database/unregister"
#define  DATACENTER_DEV_UNREGISTER_SUB "database/action/response/pemd/unregister"

//数据上报
#define  DATACENTER_ENGINE_REPORT(model,dev) "pemd/notify/event/database/"+model+"/"+dev
#define  DATACENTER_ENGINE_REPORT_SUB(model,dev) "database/notify/response/pemd/"+model+"/"+dev
//数据变化上报
#define  DATACENTER_ENGINE_DATACHANGE_REPORT(model,dev) "pemd/notify/spont/*/"+model+"/"+dev

//实时数据查询
#define  DATACENTER_REALTIME_QUERY_PUB      "pemd/get/request/database/realtime"
#define  DATACENTER_REALTIME_QUERY_SUB      "database/get/response/pemd/realtime"

//遥控命令转发
#define  DATACENTER_TRANS_REMOTER_PUB  "module104/action/request/*/remoteCtrl"
#define  DATACENTER_TRANS_REMOTER_SUB  "pemd/action/response/module104/remoteCtrl"

//定值
#define  DATACENTER_NOTIFY_SET_VALUE "database/notify/spont/allApp/parameter/+"
#define  DATACENTER_QUERY_SET_VALUE_PUB "pemd/get/request/database/parameter"
#define  DATACENTER_QUERY_SET_VALUE_SUB "database/get/response/pemd/parameter"
#define  DATACENTER_SET_VALUE_SUB "+/set/request/database/parameter"


//串口名(地址)
#define  SERIAL_PORT1_NAME  "/dev/ttyRS0"
#define  SERIAL_PORT2_NAME  "/dev/ttyRS1"
#define  SERIAL_PORT3_NAME  "/dev/ttyRS2"
#define  SERIAL_PORT4_NAME  "/dev/ttyRS3"
#define  SERIAL_PORT5_NAME  "/dev/ttyRS4"

using namespace std;


typedef enum _DC_MODBUS_DATA_TYPE{
    MODBUS_DATA_TYPE_SHORT           = 1,  //short类型
    MODBUS_DATA_TYPE_UNSIGNED_SHORT  = 2,  //无符号short类型
    MODBUS_DATA_TYPE_INT             = 3,  //int类型
    MODBUS_DATA_TYPE_UNSIGNED_INT    = 4,  //无符号int类型
    MODBUS_DATA_TYPE_FLOAT           = 5,  //浮点型
    MODBUS_DATA_TYPE_BIT             = 6,  //位
    MODBUS_DATA_TYPE_BITS            = 7   //多位
}DC_MODBUS_DATA_TYPE;

typedef enum _DC_DLT698_DATA_TYPE{
    DLT698_DATA_TYPE_SHORT           = 262,  //short类型
    DLT698_DATA_TYPE_UNSIGNED_SHORT  = 263,  //无符号short类型
    DLT698_DATA_TYPE_INT             = 264,  //int类型
    DLT698_DATA_TYPE_UNSIGNED_INT    = 265,  //无符号int类型
    DLT698_DATA_TYPE_FLOAT           = 268,  //浮点型
    DLT698_DATA_TYPE_BIT             = 260,  //位
    DLT698_DATA_TYPE_BITS            = 260   //多位
}DC_DLT698_DATA_TYPE;

//沧州项目补丁控制
bool     g_cangzhou_patch_enable = false;

DataCenterManager  DataCenterManager::m_datacenterMana;

DataCenterManager::DataCenterManager()
{
    m_req_id = 0;
    m_is_receiving = false;
    m_rsp_len = 0;

}

DataCenterManager::~DataCenterManager(){

}

DataCenterManager* DataCenterManager::getInstance()
{
    return &m_datacenterMana;
}

void DataCenterManager::init(std::shared_ptr<CConfigManager> cm, std::shared_ptr<TaskManager> tm)
{
    m_cm = cm;
    m_tm = tm;
    
    init_mqtt();
    m_bsubscribed_end = false;
}

void DataCenterManager::init_mqtt()
{
    m_mosquitto_manager = std::make_shared<CMosquittoManager>();
    m_mosquitto_manager->init(this, datacenter_callback);

}

int DataCenterManager::connect(const char *host, int port, const char *username, const char *password, int keepalive){

    if ((NULL == host) || (1 > strlen(host))){
        return MOSQ_ERR_INVAL;
    }
    if (0 > keepalive){
        keepalive = 0;
    }
        
    int ret = m_mosquitto_manager->connect(host, port, username, password, keepalive);
    if (MOSQ_ERR_SUCCESS == ret){
        m_mosquitto_manager->start();
    }
    else{
        m_mosquitto_manager->stop();
        sleep(2);
        connect(host, port, username, password, keepalive);
    }
    CLogger::getInstance()->log_i("datacenter connect host=%s,port=%d,ret=%d",host,port,ret);

    return MOSQ_ERR_SUCCESS;
}

//MQTT回调函数
int DataCenterManager::datacenter_callback(CALLBACK_TYPE cb_type, void *data, int data_len, int result, void *arg)
{
    CLogger::getInstance()->log_d("DataCenterManager: datacenter_callback(%d)", cb_type);
    if (NULL == data){
        return -1;
    }
    try{
        DataCenterManager *dcm = NULL;
        if (NULL != arg){
            dcm = (DataCenterManager *)arg;
        }
        switch (cb_type){
        case CALLBACK_TYPE_CONNECTED:
            dcm->subscribe_topics();
            dcm->m_bsubscribed_end = true;
            break;
        case CALLBACK_TYPE_MESSAGE:
            dcm->handle_message(data, data_len);
            break;
        case CALLBACK_TYPE_PUBLISH:
            break;
        case CALLBACK_TYPE_SUBSCRIBE:
            break;
        case CALLBACK_TYPE_UNSUBSCRIBE:
            break;
        case CALLBACK_TYPE_LOG:
            break;
        case CALLBACK_TYPE_DISCONNECTED:
            break;
        default:
            break;
        }
    }
    catch (...){

    }
    CLogger::getInstance()->log_d("%s{%d}", (char *)data, result);

    return result;
}


int DataCenterManager::handle_message(void *data, int data_len)
{
    if ((nullptr == data) || (0 >= data_len))
    {
        return 0;
    }

    struct mosquitto_message *msg = (struct mosquitto_message *)data;
    std::string  strtopic = msg->topic;
    std::string payload = (char *)msg->payload;
    std::string strstatus;
    CLogger::getInstance()->log_i("recv rsp msg topic: %s", strtopic.c_str());
    CLogger::getInstance()->log_i("DataCenterManager,recv rsp msg:%s", msg->payload);

    //模型设置响应、设备注册响应、设备注销响应、模型删除响应
    if ((0 == strtopic.compare(DATACENTER_MODEL_SET_SUB)) ||
         (0 == strtopic.compare(DATACENTER_DEV_REGISTER_SUB)) ||
         (0 == strtopic.compare(DATACENTER_DEV_UNREGISTER_SUB))||
         (0 == strtopic.compare(DATACENTER_MODEL_DELE_SUB)))
    {
        m_dpm.get_dc_comm_rsp(payload, strstatus);

        m_rsp_len = strstatus.length();
        m_status = strstatus;
        CLogger::getInstance()->log_i("[%s]rsp status: %s", strtopic.c_str(), strstatus.c_str());

        //发送收到数据信号
        finish_signal();

        return strstatus.length();
    }

    //设备注册通知
    if (0 == strtopic.compare(DATACENTER_DEV_REGISTER_NOTIFY_SUB)){
        m_lstRegisterNotify.clear();
        int ret = m_dpm.get_dev_register_notify(payload, m_lstRegisterNotify);
        //发送收到数据信号
        //finish_signal();
        if ((0 < ret) && (0 <m_lstRegisterNotify.size()) )
        {
            return m_lstRegisterNotify.size();
        }
    }

    //模型名称查询返回
    if (0 == strtopic.compare(DATACENTER_MODEL_NAME_QUERY_SUB))
    {
        int ret = m_dpm.get_model_name_query_rsp(payload, m_lstModelNames);
        finish_signal();
        if (0 < ret)
        {
            return m_lstModelNames.size();
        }
    }
    //模型查询返回
    if (0 == strtopic.compare(DATACENTER_MODEL_QUERY_SUB))
    {
        m_lstModels.clear();
        int ret = m_dpm.get_model_query_response(payload, m_lstModels);
        finish_signal();
        if (0 < ret)
        {
            return m_lstModels.size();
        }
    }

    //注册关系查询返回
    if (0 == strtopic.compare(DATACENTER_DEV_REGISTER_RELATION_SUB))
    {
        m_lstRegisterRelation.clear();
        int ret = m_dpm.get_model_register_relationship_rsp(payload, m_lstRegisterRelation);
        finish_signal();
        if (0 < ret)
        {
            subscribe_report_rsptopics(m_lstRegisterRelation);
            save_dev_guid(m_lstRegisterRelation);

            return m_lstRegisterRelation.size();
        }
    }

    //实时数据查询返回
    if (0 == strtopic.compare(DATACENTER_REALTIME_QUERY_SUB))
    {
        m_lstRealtimeQueryRsp.clear();
        int ret = m_dpm.get_realtime_data_rsp_data(payload, m_lstRealtimeQueryRsp);
        finish_signal();
        if (0 <ret)
        {
            return m_lstRealtimeQueryRsp.size();
        }
    } else if(true == std::regex_match(msg->topic, std::regex("database/notify/spont/allApp/parameter/(\\w+)"))){
        
        // getInstance()->handle_const_value(payload);
    }  else if(0 == strtopic.compare(DATACENTER_QUERY_SET_VALUE_SUB)){
        // finish_signal();
        // getInstance()->set_const_value(payload);
    } else if(true == std::regex_match(msg->topic, std::regex("(\\w+)/set/request/database/parameter"))){
        getInstance()->set_const_value(payload);
    }

    //数据上报响应解析
    for (auto itdev = m_vtDevGuid.begin(); itdev != m_vtDevGuid.end(); itdev++){
        std::string  strdDevguid = *itdev;
        if (strtopic.find(strdDevguid) != std::string::npos)
        {
            CLogger::getInstance()->log_d("DataCenterManager,recv datareport rsp.");
            //发送收到数据信号
            finish_signal();
            break;
        }
    }    

    return 0;
}

int DataCenterManager::subscribe_topics()
{
    //设备注册响应主题
    std::string topic = DATACENTER_DEV_REGISTER_SUB;
    int mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //模型设置响应主题
    topic = DATACENTER_MODEL_SET_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //设备注册通知主题
    topic = DATACENTER_DEV_REGISTER_NOTIFY_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //模型删除响应主题
    topic = DATACENTER_MODEL_DELE_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //设备取消响应主题
    topic = DATACENTER_DEV_UNREGISTER_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //模型名称查询、模型查询
    topic = DATACENTER_MODEL_NAME_QUERY_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }
    topic = DATACENTER_MODEL_QUERY_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }    

    //注册关系查询
    topic = DATACENTER_DEV_REGISTER_RELATION_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS)
    {
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //实时数据查询(按设备)主题
    topic = DATACENTER_REALTIME_QUERY_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS)
    {
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //控制命令转发
    topic = DATACENTER_TRANS_REMOTER_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS)
    {
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }
    //定值设置
    topic = DATACENTER_SET_VALUE_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS)
    {
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }
    //定值通知
    topic = DATACENTER_NOTIFY_SET_VALUE;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS)
    {
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }
    //定值查询
    topic = DATACENTER_QUERY_SET_VALUE_SUB;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS)
    {
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    CLogger::getInstance()->log_d("all topic register to datacenter.");

    return MOSQ_ERR_SUCCESS;
}

int DataCenterManager::subscribe_report_rsptopics(std::list<ST_REGISTER_MODEL_INFO>& lstRegisterRelation)
{
    std::string strmodel, strdev;
    int mossub =0;
    m_vtDevGuid.clear();

    for (auto itRegNotify = lstRegisterRelation.begin(); itRegNotify != lstRegisterRelation.end(); itRegNotify++)
    {
        ST_REGISTER_MODEL_INFO  stRegisterRela = *itRegNotify;
        strmodel = stRegisterRela.model;

        for (auto itd = stRegisterRela.lstRegModels.begin(); itd != stRegisterRela.lstRegModels.end(); itd++)
        {  
            strdev = itd->dev;
            m_vtDevGuid.push_back(strdev);

            std::string topic = DATACENTER_ENGINE_REPORT_SUB(strmodel, strdev);
            mossub = m_mosquitto_manager->subscribe(topic.c_str());
            if (mossub != MOSQ_ERR_SUCCESS){
                CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
                return mossub;
            }

            CLogger::getInstance()->log_d(" data report topic[%s] subscribed.", topic.c_str());
        }
    }
    
    return MOSQ_ERR_SUCCESS;
}

std::list<std::string>& DataCenterManager::dc_modelname_query_waitsubscribe()
{
    pthread_create(&m_pthread_mn_query_id, NULL, handle_modelname_query_thread, this);
    pthread_join(m_pthread_mn_query_id, NULL);
    return m_lstModelNames;
}

void  DataCenterManager::devregister_modelset_todc()
{
    pthread_create(&m_pthread_id, NULL, handle_thread, this);
    pthread_join(m_pthread_id, NULL);
}

void* DataCenterManager::handle_thread(void *arg)
{
    if (NULL == arg){
        return NULL;
    }
    DataCenterManager *dcm = (DataCenterManager *)arg;
    pid_t tid = syscall(SYS_gettid);
    printf("DataCenterManager::handle_thread------------------------------>threadid=%d\n", tid);

    //延时，等待订阅主题结束
    //先把设备取消注册，再先进行模型设置、再设备注册  
    std::this_thread::sleep_for(std::chrono::milliseconds(2000));

    std::list<std::string> lstModelNames = dcm->dc_model_name_query();
    if (lstModelNames.size() > 0) {
        dcm->dc_model_delete(lstModelNames);
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    }
    dcm->devunreg_from_datacenter();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    dcm->modelset_to_datacenter();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    dcm->devregister_to_datacenter();
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    return NULL;
}

void* DataCenterManager::handle_modelname_query_thread(void *arg)
{
    if (NULL == arg){
        return NULL;
    }
    DataCenterManager *dcm = (DataCenterManager *)arg;

    int num = 0;
    while (!dcm->m_bsubscribed_end)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        num++;
        if (num >100)
        {
            break;
        }
    }
    if (dcm->m_bsubscribed_end)
    {        
        dcm->dc_model_name_query();
    }
}

int DataCenterManager::dc_dev_register(const std::list<ST_MODEL_INFO>& lstDevinfo)
{
    m_thread_notify.lock();
    m_lstRegisterNotify.clear();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_DEV_REGISTER_PUB;

    m_dpm.get_dev_register_msg(m_req_id, lstDevinfo, msg);
    CLogger::getInstance()->log_i("dc_dev_register payload: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
    }
    else {
        CLogger::getInstance()->LOG_E("datacenter dc_dev_register [%s] error=%d.", topic.c_str(), ret);   
    }
    m_thread_notify.unlock();
    
    return 0;
}

int DataCenterManager::dc_dev_unreg(const std::list<ST_UNREG_DEVINFO>& lstDevinfo)
{
    m_thread_notify.lock();
    m_lstRegisterNotify.clear();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_DEV_UNREGISTER_PUB;

    m_dpm.get_dev_unreg_msg(m_req_id, lstDevinfo, msg);
    CLogger::getInstance()->log_d("dc_dev_unreg payload: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
    }
    else {
        CLogger::getInstance()->LOG_E("datacenter dc_dev_unreg [%s] error=%d.", topic.c_str(), ret);
    }
    m_thread_notify.unlock();

    return 0;
}

int  DataCenterManager::dc_model_set(const std::string& strmodel, std::list<ST_MODEL_PTINFO>& lstModels, std::string& status)
{
    m_thread_notify.lock();
    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_MODEL_SET_PUB;

    m_dpm.get_model_set_msg(m_req_id, strmodel, lstModels, msg);
    CLogger::getInstance()->log_i("模型设置发送数据: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();

        status = m_status;
    }
    else{
        CLogger::getInstance()->LOG_E("datacenter dc_model_set [%s] error=%d.", topic.c_str(), ret);
        m_thread_notify.unlock();

        return 0;
    }
    m_thread_notify.unlock();

    return status.length();
}

int  DataCenterManager::dc_datareport(const std::string& strmodel, const std::string& strdev, std::list<ST_REPORT_DATAPT>& lstReportPts)
{
    m_thread_notify.lock();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_ENGINE_REPORT(strmodel, strdev);

    m_dpm.get_datareport_msg(m_req_id, lstReportPts, msg);
    CLogger::getInstance()->LOG_I("dc_datareport: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();

        CLogger::getInstance()->log_i("dc_datareport,status=%s",m_status.c_str());
        //status = m_status;
    }
    else{
        CLogger::getInstance()->LOG_E("datacenter dc_datareport [%s] error=%d.", topic.c_str(), ret);
        m_thread_notify.unlock();

        return 0;
    }
    m_thread_notify.unlock();

    return m_status.length();

}

void DataCenterManager::devregister_to_datacenter()
{
    std::list<ST_MODEL_INFO> lstModelInfo;
    get_dev_reginfo(lstModelInfo);

    CLogger::getInstance()->log_d("dev register start, devnum=%d",lstModelInfo.size());
    if (lstModelInfo.size() >0)
    {
        dc_dev_register(lstModelInfo);
    }
}

int DataCenterManager::get_dev_reginfo(std::list<ST_MODEL_INFO>& lstModelInfo)
{
    auto devs = m_cm->get_terminal_devices();
    for (auto itdev = devs->begin(); itdev != devs->end(); itdev++)
    {
        int64_t rule_id = itdev->second.protocol_id;
        auto rule = m_cm->get_protocol_rule(rule_id);
        if (nullptr == rule)  {
            CLogger::getInstance()->log_d("get_dev_reginfo:get protocol rule error.");
            continue;
        }

        auto product = m_cm->get_term_product(itdev->second.terminal_product_id);
        if (nullptr == product) {
            CLogger::getInstance()->log_d("get_dev_reginfo:get terminal product error.");
            continue;
        }
        

        ST_MODEL_INFO  stModelInfo;
        stModelInfo.model = product->model_name;
        if (itdev->second.parent_id > 0){
            stModelInfo.addr = get_comp_addr(itdev->second.slave_addr, itdev->second.parent_id);
        }
        else{
            stModelInfo.addr = itdev->second.slave_addr;
        }

        stModelInfo.productID = std::to_string(itdev->second.id);
        stModelInfo.port = get_devport(itdev->second.channel);
        stModelInfo.isreport = "1";
        stModelInfo.desc = itdev->second.desc;
        stModelInfo.nodeID = itdev->second.node_id;
        if (product)
        {
            stModelInfo.manuID = product->manuId;
        }
        else{
            stModelInfo.manuID = "1";   //若无产品信息，置常量1
        }
        
        lstModelInfo.push_back(stModelInfo);

        //冻结数据模型单独注册
        auto itfroz = m_mpFrozenItems.find(rule_id);
        if (itfroz != m_mpFrozenItems.end())
        {
            std::string  frozen_model = product->model_name + "_frozen";
            stModelInfo.model = frozen_model;

            stModelInfo.nodeID = itdev->second.node_id;
            stModelInfo.desc = itdev->second.desc;
            if (itdev->second.parent_id > 0)
            {
                stModelInfo.addr = get_comp_addr(itdev->second.node_id, itdev->second.parent_id);
            }
            else{
                stModelInfo.addr = itdev->second.node_id;
            }
            lstModelInfo.push_back(stModelInfo);
        }
    }

    return lstModelInfo.size();
}

void DataCenterManager::devunreg_from_datacenter()
{
    std::list<ST_UNREG_DEVINFO> lstModelInfo;

    auto devguids = DataCenterManager::getInstance()->get_all_devguids();
    for (auto itdev = devguids.begin(); itdev != devguids.end(); itdev++)
    {
        ST_UNREG_DEVINFO  stModelInfo;

        stModelInfo.model = itdev->model;
        stModelInfo.addr = itdev->addr;
        stModelInfo.port = itdev->port;
        stModelInfo.desc = itdev->description;
        
        lstModelInfo.push_back(stModelInfo);
    }

    CLogger::getInstance()->log_d("dev unregister start: devnum=%d", devguids.size());
    if (lstModelInfo.size() >0)
    {
        dc_dev_unreg(lstModelInfo);
    }   
}

void DataCenterManager::modelset_to_datacenter()
{
    std::unordered_map<std::string, int>   model_modbus;
    std::unordered_map<std::string, int>   model_dlt645;
    std::unordered_map<std::string, int>   model_dlt698;

    get_dlt645_frozenitems();

    auto devs = m_cm->get_terminal_devices();
    for (auto itm = devs->begin(); itm != devs->end(); itm++)
    {       
        auto proto_rule = m_cm->get_protocol_rule(itm->second.protocol_id);
        if (nullptr == proto_rule) {
            CLogger::getInstance()->log_d("modelset_to_datacenter:query protocol rule error.");
            continue;
        }

        auto product = m_cm->get_term_product(itm->second.terminal_product_id);
        if (nullptr == product) {
            CLogger::getInstance()->log_d("modelset_to_datacenter:query terminal product error.");
            continue;
        }

        std::string model_name = product->model_name;
        std::string frozen_mode = model_name + "_frozen";

        int64_t ruleid = proto_rule->id;
        int rultype =proto_rule->protocol_type ;   //1:modbus,2:645,3:698
        
        if (1== rultype){
            auto itfind = model_modbus.find(model_name);
            if (itfind == model_modbus.end())  {
                dc_model_mobusset(ruleid, model_name);
                model_modbus.insert(std::make_pair(model_name, 1));
            }            
        }
        else if (2 == rultype) {
            auto itfind = model_dlt645.find(model_name);
            if (itfind == model_dlt645.end())  {
                dc_model_dlt645set(ruleid, model_name);
                model_dlt645.insert(std::make_pair(model_name, 1));
                //冻结数据模型单独注册
                if (m_mpFrozenItems.size() >0)
                {
                    dc_model_dlt645set(ruleid, frozen_mode,true);
                    model_dlt645.insert(std::make_pair(frozen_mode, 1));
                }
            }
        }
        else if (3 == rultype)
        {
            auto itfind = model_dlt698.find(model_name);
            if (itfind == model_dlt698.end())
            {
                dc_model_dlt698set(ruleid, model_name);
                model_dlt698.insert(std::make_pair(model_name, 1));
            }           
        }
    }
    
    CLogger::getInstance()->log_d("model set end,dev num=%d",devs->size());

}

void DataCenterManager::dc_model_mobusset(int64_t protocol_id, std::string model)
{
    std::list<ST_MODEL_PTINFO> lstModelPts;
    std::string status;

    ST_MODEL_PTINFO  stModelPt;
    //modbus采集点
    auto modbus_pts = m_cm->get_modbus_points();
    for (auto itmod = modbus_pts->begin(); itmod != modbus_pts->end(); itmod++)
    {
        if (itmod->second.protocol_id == protocol_id)
        {
            if (itmod->second.name.empty())
            {
                CLogger::getInstance()->LOG_E("dc_model_mobusset:modbus request point name is empty.id=%lld", itmod->second.id);
                continue;
            }
            stModelPt.name = itmod->second.name;
            stModelPt.ratio = std::to_string(itmod->second.scale);
            stModelPt.type = get_modbus_datatype(itmod->second.data_type);
            if (itmod->second.unit.empty()) {
                stModelPt.unit = " ";
            }
            else{
                stModelPt.unit = itmod->second.unit;
            }
            
            stModelPt.isreport = "1";
            stModelPt.deadzone = std::to_string(itmod->second.lower_limit);   //toDo: ??
            stModelPt.userdefine = itmod->second.description;

            lstModelPts.push_back(stModelPt);
        }
    }
    if (g_cangzhou_patch_enable)
    {
        CCangZhouRptPatch::getInstance()->AppendModelItem(lstModelPts);
        CLogger::getInstance()->LOG_I("dc_model_mobusset:cangzhou report item appended.");
    }
    //modbus命令点
    auto modbus_cmds = m_cm->get_modbus_commands();
    for (auto itcmd = modbus_cmds->begin(); itcmd != modbus_cmds->end(); itcmd++)
    {
        if (itcmd->second.protocol_id == protocol_id)
        {
            if (itcmd->second.name.empty())
            {
                CLogger::getInstance()->LOG_E("dc_model_mobusset:modbus command point name is empty.id=%lld", itcmd->second.id);
                continue;
            }
            stModelPt.name = itcmd->second.name;
            stModelPt.ratio = "1";  //
            stModelPt.type = get_modbus_datatype(itcmd->second.data_type);
            stModelPt.unit = " ";         //单位设为空格
            stModelPt.isreport = "0";    //命令，不上报
            stModelPt.deadzone = "0";    //死区，默认设置成0
            stModelPt.userdefine = itcmd->second.description;

            lstModelPts.push_back(stModelPt);
        }
    }
    if (lstModelPts.size() > 0)
    {
        dc_model_set(model, lstModelPts, status);
        CLogger::getInstance()->LOG_I("dc_model_mobusset model[%s] set %s", model.c_str(), status.c_str());
    }
    else{
        CLogger::getInstance()->log_d("no modbus points to register to datacenter.");
    }
    
}

void DataCenterManager::dc_model_dlt645set(int64_t protocol_id, std::string model, bool bfrozen)
{
    std::list<ST_MODEL_PTINFO> lstModelPts;
    std::string status;

    ST_MODEL_PTINFO  stModelPt;
    //645采集点
    auto dlt645_pts = m_cm->get_dlt645_points();
    for (auto itd = dlt645_pts->begin(); itd != dlt645_pts->end(); itd++)
    {
        if (!bfrozen)
        {
            auto itp = m_mpFrozenItems.find(protocol_id);
            if (itp != m_mpFrozenItems.end())
            {
                auto itf = itp->second.find(itd->second.di);
                if (itf != itp->second.end())
                {
                    continue;    //当前点在冻结请求中，实时数据模型不处理
                }
            }
        }
        else{
            auto itp = m_mpFrozenItems.find(protocol_id);
            if (itp == m_mpFrozenItems.end()) {
                continue;   //当前协议不在冻结数据请求中
            }
            else{
                auto itf = itp->second.find(itd->second.di);
                if (itf == itp->second.end())  {
                    continue;    //当前点不在冻结请求中，冻结数据模型不处理
                }
            }
        }

        if (itd->second.protocol_id == protocol_id)
        {
            if (itd->second.name.empty())
            {
                CLogger::getInstance()->LOG_E("dc_model_dlt645set:645 request point name is empty.id=%lld", itd->second.id);
                continue;
            }
            stModelPt.name = itd->second.name;
            stModelPt.ratio = std::to_string(itd->second.scale);
            stModelPt.type = get_dlt645_datatype(itd->second.data_type, itd->second.decimal);
            if (itd->second.unit.empty()) {
                stModelPt.unit = " ";
            }
            else{
                stModelPt.unit = itd->second.unit;
            }
            stModelPt.isreport = "1";
            stModelPt.deadzone = std::to_string(itd->second.lower_limit);   //toDo: ??
            stModelPt.userdefine = itd->second.description;

            lstModelPts.push_back(stModelPt);
        }
    }
    //645命令点
    auto dlt645_cmds = m_cm->get_dlt645_commands();
    for (auto itd = dlt645_cmds->begin(); itd != dlt645_cmds->end(); itd++)
    {
        if (itd->second.protocol_id == protocol_id)
        {
            if (itd->second.name.empty())
            {
                CLogger::getInstance()->LOG_E("dc_model_dlt645set:645 command point name is empty.id=%lld",itd->second.id);
                continue;
            }
            stModelPt.name = itd->second.name;
            stModelPt.ratio = "1";
            stModelPt.type = "Int";   //数据类型设置为空
            stModelPt.unit = " ";    //单位默认设置为空格
            stModelPt.isreport = "0";  //不上报
            stModelPt.deadzone = "0";  //死区设置为0
            stModelPt.userdefine = itd->second.description;

            lstModelPts.push_back(stModelPt);
        }
    }
    if (lstModelPts.size() >0)
    {
        dc_model_set(model, lstModelPts, status);
        CLogger::getInstance()->log_d("dc_model_dlt645set model[%s] set %s", model.c_str(), status.c_str());
    }
    else{
        CLogger::getInstance()->log_d("no dlt645 points to register to datacenter.");
    }
    
}

void DataCenterManager::dc_model_dlt698set(int64_t protocol_id, std::string model)
{
    std::list<ST_MODEL_PTINFO> lstModelPts;
    std::string status;

    ST_MODEL_PTINFO  stModelPt;
    auto dlt698_pts = m_cm->get_dlt698_points();
    for (auto itd = dlt698_pts->begin(); itd != dlt698_pts->end(); itd++)
    {
        if (itd->second.protocol_id == protocol_id)
        {
            if (itd->second.name.empty())
            {
                CLogger::getInstance()->LOG_E("dc_model_dlt698set:dlt698 request point name is empty.id=%lld", itd->second.id);
                continue;
            }
            stModelPt.name = itd->second.name;
            stModelPt.ratio = std::to_string(itd->second.scale);
            stModelPt.type = get_dlt698_datatype(itd->second.data_type);
            stModelPt.unit = itd->second.unit;
            stModelPt.isreport = "1";
            stModelPt.deadzone = std::to_string(itd->second.lower_limit);   //toDo: ??
            stModelPt.userdefine = itd->second.description;

            lstModelPts.push_back(stModelPt);
        }
    }
    if (lstModelPts.size() > 0)
    {
        dc_model_set(model, lstModelPts, status);
        CLogger::getInstance()->LOG_I("dc_model_dlt698set model[%s] set %s", model.c_str(), status.c_str());
    }
    else{
        CLogger::getInstance()->log_d("no dlt698 points to register to datacenter.");
    }
}

void DataCenterManager::finish_signal(){
    m_thread_notify.lock();
    m_is_receiving = false;
    m_thread_notify.signal();
    m_thread_notify.unlock();
}

void DataCenterManager::wait_recv_finished(){
    //最长等待 1S *10
    int ms = 1000;
    int n = 0;
    while (m_is_receiving){
        n++;
        if (n > 10){
            break;
        }
        m_thread_notify.timewait(ms);
    }
    if (n > 10){
        m_rsp_len = 0;
    }
}

std::string  DataCenterManager::get_devport(std::string& channel)
{
    std::string  port ="HPLC";
    if (channel.compare(SERIAL_PORT1_NAME) == 0) {
        port = "RS485-1";
    }
    else if (channel.compare(SERIAL_PORT2_NAME) == 0){
        port = "RS485-2";
    }
    else if (channel.compare(SERIAL_PORT3_NAME) == 0){
        port = "RS485-3";
    }
    else if (channel.compare(SERIAL_PORT4_NAME) == 0){
        port = "RS485-4";
    }
    else if (channel.compare(port.c_str()) == 0){
        port = "HPLC";
    }

    return port;
}

std::string DataCenterManager::get_modbus_datatype(int data_type)
{
    std::string str_datatype;
    switch (data_type)
    {
    case DATA_TYPE_SHORT:
    case DATA_TYPE_UNSIGNED_SHORT:
    case DATA_TYPE_INT:
    case DATA_TYPE_UNSIGNED_INT:
    case DATA_TYPE_FLOAT:
        str_datatype = "float";
        break;
    case DATA_TYPE_BIT:
    default:
        str_datatype = "int";
        break;
    }

    return str_datatype;
}

std::string DataCenterManager::get_dlt645_datatype(int data_type, int decimmal)
{
    std::string  str_datatype;
    int dataClass = 1;   //1:数值型，2：字符串
    auto  datatypes = m_cm->get_dlt645_data_type();
    for (auto itTy = datatypes->begin(); itTy != datatypes->end(); itTy++)
    {
        if (data_type == itTy->second.id)
        {
            dataClass = itTy->second.data_type;
            break;
        }
    }

    if ((1 ==dataClass) && (decimmal >0) )
    {
        str_datatype = "float";
    }else if ((1 == dataClass) && (decimmal ==0))
    {
        str_datatype = "int";
    }
    else{
        str_datatype = "string";
    }

    return str_datatype;
}

std::string DataCenterManager::get_dlt698_datatype(int data_type)
{
    std::string str_datatype = "int";
    auto dlt698_types = m_cm->get_dlt698_data_type();
    for (auto itt = dlt698_types->begin(); itt != dlt698_types->end(); itt++)
    {
        if (itt->second.code == data_type)
        {
            str_datatype = itt->second.name;
            break;
        }
    }

    return str_datatype;
}

int DataCenterManager::save_dev_guid(std::list<ST_REGISTER_MODEL_INFO>& lstRegisterRelation)
{    
    CLogger::getInstance()->log_d("save_dev_guid:size=%d", lstRegisterRelation.size());

    int num = 0;
    for (auto itReg = lstRegisterRelation.begin(); itReg != lstRegisterRelation.end(); itReg++)
    {
        std::string model = itReg->model;
        std::string port = itReg->port;

        for (auto itm = itReg->lstRegModels.begin(); itm != itReg->lstRegModels.end(); itm++)
        {
            ST_DEV_GUID stDevguid;
            stDevguid.id = atoll(itm->productID.c_str());    //注册时，产品ID使用的是dev_id
            stDevguid.dev = itm->dev;
            stDevguid.model = model;
            stDevguid.port = port;
            stDevguid.guid = itm->guid;
            stDevguid.addr = itm->addr;
            stDevguid.description = itm->desc;

            std::string key = get_dc_devkey(model, port, itm->addr, itm->desc);
            auto itID = m_mpModeldesc.find(key);
            if (itID == m_mpModeldesc.end())
            {       
                CDBManager::getInstance()->insert_dev_guid(stDevguid);
                m_mpModeldesc.insert(std::make_pair(key, stDevguid.id));
                //暂存到CConfigManger的map中，后面使用时查询使用，不再查询数据库
                std::string padr = stDevguid.port+"_"+stDevguid.addr;
                set_devguid_type1(padr, stDevguid);
                set_devguid_type2(stDevguid.dev, stDevguid);
                CLogger::getInstance()->LOG_I("save_dev_guid:key = %s, dev=%s, model=%s", key.c_str(), stDevguid.dev.c_str(), stDevguid.model.c_str());           
                num++;
            }
        }   
        
    }

    CLogger::getInstance()->log_d("save_dev_guid: num=%d", num);


    return 0;
}

std::list<std::string>& DataCenterManager::dc_model_name_query()
{
    m_thread_notify.lock();
    m_lstModelNames.clear();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_MODEL_NAME_QUERY_PUB;

    std::list<std::string>  lstNames;
    m_dpm.get_model_query_request(m_req_id, lstNames, msg);
    CLogger::getInstance()->log_d("dc_model_name_query: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
    } else {
        CLogger::getInstance()->LOG_E("dc_model_name_query error=%d.", ret);
    }
    m_thread_notify.unlock();

    return m_lstModelNames;

}

std::list<ST_SINGLE_MODEL_INFO>& DataCenterManager::dc_model_query(const std::list<std::string>& lstModelNames)
{
    m_thread_notify.lock();
    m_lstModels.clear();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_MODEL_QUERY_PUB;

    m_dpm.get_model_query_request(m_req_id, lstModelNames, msg);
    CLogger::getInstance()->log_d("dc_model_query: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
        CLogger::getInstance()->log_i("dc_model_query,status=%s", m_status.c_str());
    }
    else{
        CLogger::getInstance()->LOG_E("dc_model_query error=%d.", ret);
    }
    m_thread_notify.unlock();

    return m_lstModels;
}

void DataCenterManager::dc_model_delete(const std::list<std::string>& lstModelNames)
{
    m_thread_notify.lock();
    m_lstModels.clear();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_MODEL_DELE_PUB;

    m_dpm.get_model_dele_msg(m_req_id, lstModelNames, msg);
    CLogger::getInstance()->log_d("dc_model_delete: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();

        CLogger::getInstance()->log_i("dc_model_delete,status=%s", m_status.c_str());
    }
    else{
        CLogger::getInstance()->LOG_E("dc_model_delete error=%d.", ret);
    }
    m_thread_notify.unlock();
}

void DataCenterManager::dc_datachange_report(const std::string& strmodel, const std::string& strdev, int datatype, std::list<ST_CHANGE_REPORT_DATAPT> lstReportPts)
{
    m_thread_notify.lock();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_ENGINE_DATACHANGE_REPORT(strmodel,strdev);

    m_dpm.get_datachange_report_msg(m_req_id, datatype, lstReportPts, msg);
    CLogger::getInstance()->LOG_I("dc_datachange_report: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
    }
    else{
        CLogger::getInstance()->LOG_E("dc_model_query error=%d.", ret);
    }
    m_thread_notify.unlock();
}

int DataCenterManager::dc_register_relation_query(std::list<ST_REGISTER_MODEL_INFO>& lstRegisterRelation)
{
    m_thread_notify.lock();
    m_lstRegisterRelation.clear();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_DEV_REGISTER_RELATION_PUB;

    m_dpm.get_model_register_relationship_req(m_req_id, msg);
    CLogger::getInstance()->log_d("dc_register_relation_query: %s", msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();

        //将解析得到的复制到返回list
        get_register_relations(lstRegisterRelation);
    }
    else{
        CLogger::getInstance()->LOG_E("dc_register_relation_query error=%d.", ret);
        m_thread_notify.unlock();

        return 0;
    }
    m_thread_notify.unlock();

    return lstRegisterRelation.size();
}

int DataCenterManager::dc_transpond_remotectrl_cmd_req(std::list<ST_REMOTECTRL_CMD>& lstRemoteCtrlcmd)
{
    m_thread_notify.lock();
   
    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_TRANS_REMOTER_PUB;

    m_dpm.get_trans_remotectrl_req_msg(m_req_id, lstRemoteCtrlcmd, msg);
    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
       //数据中心无响应，无需等待，直接返回。响应由单独的接口返回
    }
    else{
        CLogger::getInstance()->LOG_E("dc_transpond_remotectrl_cmd_req error=%d.", ret);
        m_thread_notify.unlock();

        return 0;
    }
    m_thread_notify.unlock();

    return 1;
}
#if 0
int DataCenterManager::dc_transpond_remotectrl_cmd_rsp(std::list<ST_REMOTECTRL_RSP>& lstRemoteCtrlRsp)
{
    m_thread_notify.lock();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_TRANS_REMOTER_SUB;

    m_dpm.get_trans_remotectrl_rsp_msg(m_req_id, lstRemoteCtrlRsp, msg);
    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        //数据中心无响应，无需等待，直接返回
    }
    else{
        CLogger::getInstance()->LOG_E("dc_transpond_remotectrl_cmd_rsp error=%d.", ret);
        m_thread_notify.unlock();

        return 0;
    }
    m_thread_notify.unlock();

    return 1;
}
#endif

void DataCenterManager::get_query_models(std::list<ST_SINGLE_MODEL_INFO>& lstModels)
{
    for (auto itmd = m_lstModels.begin(); itmd != m_lstModels.end(); itmd++)
    {
        ST_SINGLE_MODEL_INFO stSingleModel;
        stSingleModel.model = itmd->model;
        for (auto itInfo = itmd->lstModelptInfo.begin(); itInfo != itmd->lstModelptInfo.end(); itInfo++)
        {
            stSingleModel.lstModelptInfo.push_back(*itInfo);
        }
        lstModels.push_back(stSingleModel);
    }
}

void DataCenterManager::get_query_model_names(std::list<std::string>& lstModelNames)
{
    for (auto itn = m_lstModelNames.begin(); itn != m_lstModelNames.end(); itn++)
    {
        lstModelNames.push_back(*itn);
    }
}

void DataCenterManager::get_register_relations(std::list<ST_REGISTER_MODEL_INFO>& lstRegisterRelation)
{
    CLogger::getInstance()->log_d("get_register_relations,query rsp size=%d", m_lstRegisterRelation.size());
    for (auto itR = m_lstRegisterRelation.begin(); itR != m_lstRegisterRelation.end(); itR++)
    {
        ST_REGISTER_MODEL_INFO  stRegisterInfo;
        stRegisterInfo.model = itR->model;
        stRegisterInfo.port = itR->port;
        for (auto itDev = itR->lstRegModels.begin(); itDev != itR->lstRegModels.end(); itDev++)
        {
            stRegisterInfo.lstRegModels.push_back(*itDev);
            CLogger::getInstance()->log_d("get_register_relations,model=%s,pointname=%s", itR->model.c_str(), itDev->dev.c_str());
        }

        lstRegisterRelation.push_back(stRegisterInfo);
    }
}

std::string DataCenterManager::get_dc_devkey(std::string& model, std::string& port, std::string& addr, std::string& desc)
{
    std::string key = model ;
    key.append("_");
    key += port;
    key.append("_");
    key += addr;
    key.append("_");
    key += desc;
    
    return key;
}

std::list<ST_REALTIME_QUERY_DEV_RSP>& DataCenterManager::dc_query_realtime_datas(const std::list<std::string>& lstDevs)
{
    m_thread_notify.lock();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_REALTIME_QUERY_PUB;

    m_dpm.get_realtime_data_req_msg(m_req_id, lstDevs, msg);
    CLogger::getInstance()->log_d("dc_query_realtime_datas:%s",msg.c_str());
    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
    }
    else{
        CLogger::getInstance()->LOG_E("dc_query_realtime_datas error=%d.", ret);
    }
    m_thread_notify.unlock();

    return m_lstRealtimeQueryRsp;
}
int DataCenterManager::get_mapping_devs(const std::list<int64_t>& devs, std::unordered_map<std::string, int64_t>& dev_guids){
    for(auto dev_id = devs.begin(); dev_id != devs.end(); dev_id++){
        auto dev_guid = get_devguid_with_devid(*dev_id);
        if(nullptr != dev_guid){
            dev_guids.insert(std::make_pair(dev_guid->dev, *dev_id));
        }
    }
    return dev_guids.size();
}
/*-----------------------------------------------*/
/*可移到各模块中*/
int DataCenterManager::get_modbus_point_id(MAPPING_POINT_DATA_TYPE& mapping_point){
    auto points = m_cm->get_modbus_points();
    if(nullptr == points){
        return -1;
    }
    for(auto point = points->begin(); point != points->end(); point++){
        if(0 == point->second.name.compare(mapping_point.name)){
            mapping_point.point_id = point->second.id;
            mapping_point.data_type = point->second.data_type;
            return 0;
        }
    }
    return -1;
}
/*-----------------------------------------------*/
/*可移到各模块中*/
int DataCenterManager::get_dlt645_point_id(MAPPING_POINT_DATA_TYPE& mapping_point){
    auto points = m_cm->get_dlt645_points();
    if(nullptr == points){
        return -1;
    }
    for(auto point = points->begin(); point != points->end(); point++){
        if(0 == point->second.name.compare(mapping_point.name)){
            mapping_point.point_id = point->second.id;
            mapping_point.data_type = point->second.data_type;
            return 0;
        }
    }
    return -1;
}
/*-----------------------------------------------*/
/*可移到各模块中*/
int DataCenterManager::get_dlt698_point_id(MAPPING_POINT_DATA_TYPE& mapping_point){
    auto points = m_cm->get_dlt645_points();
    if(nullptr == points){
        return -1;
    }
    for(auto point = points->begin(); point != points->end(); point++){
        if(0 == point->second.name.compare(mapping_point.name)){
            mapping_point.point_id = point->second.id;
            mapping_point.data_type = point->second.data_type;
            return 0;
        }
    }
    return -1;
}
/*-----------------------------------------------*/
/*可移到各模块中*/
int DataCenterManager::get_point_id(MAPPING_POINT_DATA_TYPE& mapping_point){

    switch(mapping_point.protocol_type){
        case 1:
            return get_modbus_point_id(mapping_point);
        break;
        case 2:
            return get_dlt645_point_id(mapping_point);
        break;
        case 3:
            return get_dlt698_point_id(mapping_point);
        break;
        default:
        break;
    }
    return -1;
}
int DataCenterManager::dc_query_realtime_datas(const std::list<int64_t>& devs, T_MAPPING_DATA_TYPE& mapping_datas){
    //获取映射设备信息
    if(devs.empty()){
        return 0;
    }
    std::unordered_map<std::string, int64_t> dev_guids;
    if(1 > get_mapping_devs(devs, dev_guids)){
        return 0;
    }
    //获取注册设备的dev列表
    std::list<std::string> dc_devs;
    for(auto dev_guid = dev_guids.begin(); dev_guid != dev_guids.end(); dev_guid++){
        dc_devs.push_back(dev_guid->first);
    }
    if(dc_devs.empty()){
        return 0;
    }
    //获取实时数据
    std::list<ST_REALTIME_QUERY_DEV_RSP>& rt_datas = dc_query_realtime_datas(dc_devs);
    if(rt_datas.empty()){
        return 0;
    }
    for(auto rt_dev_data = rt_datas.begin(); rt_dev_data != rt_datas.end(); rt_dev_data++){
        auto dev_guid = dev_guids.find(rt_dev_data->dev);
        if(dev_guid == dev_guids.end()){
            continue;
        }
        auto terminal_dev = m_cm->get_terminal_device(dev_guid->second);
        if(nullptr == terminal_dev){
            continue;
        }
        auto protocol = m_cm->get_protocol_rule(terminal_dev->protocol_id);
        if(nullptr == protocol){
            continue;
        }

        auto mapping_dev_data = mapping_datas.find(dev_guid->second);
        if(mapping_dev_data == mapping_datas.end()){
            T_MAPPING_POINT_DATA_TYPE pds;
            for(auto rt_point_data = rt_dev_data->lstQueryDataItems.begin(); rt_point_data != rt_dev_data->lstQueryDataItems.end(); rt_point_data++){
                MAPPING_POINT_DATA_TYPE pd;
                pd.name = rt_point_data->name;
                pd.protocol_type = protocol->protocol_type;
                get_point_id(pd);
                pd.value  = rt_point_data->val;
                pds.insert(std::make_pair(pd.point_id, pd));
            }
            mapping_datas.insert(std::make_pair(dev_guid->second, pds));
        } else {
            for(auto rt_point_data = rt_dev_data->lstQueryDataItems.begin(); rt_point_data != rt_dev_data->lstQueryDataItems.end(); rt_point_data++){
                MAPPING_POINT_DATA_TYPE pd;
                pd.name = rt_point_data->name;
                pd.protocol_type = protocol->protocol_type;
                get_point_id(pd);
                pd.value  = rt_point_data->val;
                auto mapping_dev_points = mapping_dev_data->second.find(pd.point_id);
                if(mapping_dev_points == mapping_dev_data->second.end()){
                    mapping_dev_data->second.insert(std::make_pair(pd.point_id, pd));
                } else {
                    mapping_dev_data->second[pd.point_id] = pd;
                }
            }
        }
    }
    return 1;
}

void DataCenterManager::get_realtime_query_rspdata(std::list<ST_REALTIME_QUERY_DEV_RSP>& lstQueryDevs)
{
    for (auto itr = m_lstRealtimeQueryRsp.begin(); itr != m_lstRealtimeQueryRsp.end(); itr++)
    {
        ST_REALTIME_QUERY_DEV_RSP  stQueryDevRsp;
        stQueryDevRsp.dev = itr->dev;
        for (auto itd = itr->lstQueryDataItems.begin(); itd != itr->lstQueryDataItems.end(); itd++)
        {
            stQueryDevRsp.lstQueryDataItems.push_back(*itd);
        }
        lstQueryDevs.push_back(stQueryDevRsp);
    }
}

int  DataCenterManager::dc_frozen_datareport(const std::string& strmodel, const std::string& strdev, std::list<ST_FROZEN_GROUP_DATA>& lstfrozen_group, std::string frozen_type)
{
    m_thread_notify.lock();

    m_req_id++;
    std::string msg = "";
    std::string topic = DATACENTER_ENGINE_REPORT(strmodel, strdev);

    m_dpm.get_frozen_datareport_msg(m_req_id, lstfrozen_group, frozen_type, msg);
    CLogger::getInstance()->log_i("dc_frozen_datareport,type=%s: %s", frozen_type.c_str(), msg.c_str());

    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();

        CLogger::getInstance()->log_i("dc_frozen_datareport,status=%s", m_status.c_str());
    }
    else{
        CLogger::getInstance()->LOG_E("datacenter dc_frozen_datareport [%s] error=%d.", topic.c_str(), ret);
        m_thread_notify.unlock();

        return 0;
    }
    m_thread_notify.unlock();

    return m_status.length();
}


std::string DataCenterManager::get_comp_addr(std::string & addr, int64_t pid, bool bfrozen)
{
    std::string  comp_addr = addr;
    auto pdev = m_cm->get_terminal_device(pid);
    if (nullptr != pdev)
    {
        char buff[128] = { 0 };
        if (!bfrozen) {
            snprintf(buff, sizeof(buff), "%s_%s", pdev->slave_addr.c_str(), addr.c_str());
        } else{
            snprintf(buff, sizeof(buff), "%s_%s", pdev->node_id.c_str(), addr.c_str());
        }     
        return std::string(buff);
    }

    return  comp_addr;
}

void  DataCenterManager::get_dlt645_frozenitems()
{
    m_mpFrozenItems.clear();

    auto frozen_reqs = m_cm->get_dlt645_freeze_requests();
    auto frozen_times = m_cm->get_dlt645_freeze_times();
    for (auto itt = frozen_times->begin(); itt != frozen_times->end(); itt++)
    {
        int fr_type = itt->second.type;
        //若为冻结数据，保存冻结时间的di
        if (1 == fr_type)  //1:冻结数据，2：负荷记录
        {
            if (!itt->second.time_or_di.empty())
            {
                auto itf = m_mpFrozenItems.find(itt->second.protocol_id);
                if (itf == m_mpFrozenItems.end())
                {
                    std::unordered_map<std::string, int>   mpItems;
                    mpItems.insert(std::make_pair(itt->second.time_or_di, fr_type));

                    m_mpFrozenItems.insert(std::make_pair(itt->second.protocol_id, mpItems));
                }
                else{
                    auto itd = itf->second.find(itt->second.time_or_di);
                    if (itd == itf->second.end())
                    {
                        itf->second.insert(std::make_pair(itt->second.time_or_di,fr_type));
                    }
                }
            }   
        }

        auto it_req = frozen_reqs->begin();
        while (it_req != frozen_reqs->end())
        {
            //与当前冻结时间的id相等，保存di
            if (itt->second.id == it_req->second.freeze_time_id)
            {
                auto itf = m_mpFrozenItems.find(itt->second.protocol_id);
                if (itf == m_mpFrozenItems.end())
                {
                    std::unordered_map<std::string, int>   mpItems;
                    mpItems.insert(std::make_pair(it_req->second.di, fr_type));

                    m_mpFrozenItems.insert(std::make_pair(itt->second.protocol_id, mpItems));
                }
                else{
                    auto itd = itf->second.find(it_req->second.di);
                    if (itd == itf->second.end())
                    {
                        itf->second.insert(std::make_pair(it_req->second.di, fr_type));
                    }
                }
            }
            it_req++;
        }

    }
}

bool DataCenterManager::model_contrast(std::list<ST_SINGLE_MODEL_INFO> lstModels/*数据中心模型查询返回*/)
{
    std::unordered_map<std::string, std::list<std::string>>  model_props;//终端设备使用的模型名和所有属性名
    auto devs = m_cm->get_terminal_devices();
    for (auto itm = devs->begin(); itm != devs->end(); itm++)
    {
        auto proto_rule = m_cm->get_protocol_rule(itm->second.protocol_id);
        if (nullptr == proto_rule) {
            CLogger::getInstance()->log_d("model_contrast:protocol rule query error.");
            continue;
        }
        auto product = m_cm->get_term_product(itm->second.terminal_product_id);
        if(nullptr == product){
            CLogger::getInstance()->log_d("model_contrast:terminal product error.");
            continue;
        }
        auto itmodel = model_props.find(product->model_name);
        if (itmodel != model_props.end()) {
            continue;  //模型名重复，继续下一个
        }

        std::list<std::string> lstModelProps;
        int64_t protocol_id = proto_rule->id;
        int rultype = proto_rule->protocol_type;   //1:modbus,2:645,3:698
        if (1 == rultype){
            auto modbus_pts = m_cm->get_modbus_points();
            for (auto itmod = modbus_pts->begin(); itmod != modbus_pts->end(); itmod++) {
                if (itmod->second.protocol_id == protocol_id) {
                    lstModelProps.push_back(itmod->second.name);
                }
            }
        }else if (2 ==rultype) {
            auto dlt645_pts = m_cm->get_dlt645_points();
            for (auto it645 = dlt645_pts->begin(); it645 != dlt645_pts->end(); it645++)
            {
                if (it645->second.protocol_id == protocol_id)  {
                    lstModelProps.push_back(it645->second.name);
                }
            }

        }else if (3 ==rultype)  {
            auto dlt698_pts = m_cm->get_dlt698_points();
            for (auto it698 = dlt698_pts->begin(); it698 != dlt698_pts->end(); it698++)
            {
                if (it698->second.protocol_id ==protocol_id) {
                    lstModelProps.push_back(it698->second.name);
                }
            }
        }

        model_props.insert(std::make_pair(product->model_name, lstModelProps));
    }

    for (auto itm = model_props.begin(); itm != model_props.end(); itm++)
    {
        bool  same_model = false;
        std::string model = itm->first;
        for (auto itq = lstModels.begin(); itq != lstModels.end(); itq++)
        {
            if (model.compare(itq->model) == 0){
                same_model = true;
                for (auto itprop1 = itm->second.begin(); itprop1 != itm->second.end(); itprop1 ++)
                {
                    bool cur_same = false;
                    std::string prop_name = *itprop1;
                    for (auto itprop2 = itq->lstModelptInfo.begin(); itprop2 != itq->lstModelptInfo.end(); itprop2++)
                    {
                        if (prop_name.compare(itprop2->name) ==0) {
                            cur_same = true;
                            break;
                        }
                    }
                    if (!cur_same) {
                        return false;   //属性名在数据中心找不到
                    }
                }
            }
        }
        if (!same_model) {
            return false;   //模型名在数字中心找不到
        }
    }
    
    return true;
}

//对比model、port和addr
//三者都相等认为该注册设备已被注册
bool  DataCenterManager::regdev_contrast(std::list<ST_REGISTER_MODEL_INFO> lstRegisterRelation/*数据中心查询返回*/)
{
    if (0 ==lstRegisterRelation.size())
    {
        return false;
    }

    get_dlt645_frozenitems();

    std::list<ST_MODEL_INFO> lstModelInfo;
    get_dev_reginfo(lstModelInfo);

    for (auto itm = lstModelInfo.begin(); itm != lstModelInfo.end(); itm++)
    {
        bool bfind = false;
        std::string  model= itm->model;
        std::string  port = itm->port;
        std::string  addr = itm->addr;
        for (auto itr = lstRegisterRelation.begin(); itr != lstRegisterRelation.end(); itr++)
        {
            if ((itr->model.compare(model) ==0) && (itr->port.compare(port) ==0))
            {
                for (auto itd = itr->lstRegModels.begin(); itd != itr->lstRegModels.end(); itd++)
                {
                    if (itd->addr.compare(addr) ==0)
                    {
                        bfind = true;
                        break;
                    }
                }  
                if (bfind){
                    break;
                }
            }
            if (bfind) {
                break;
            }
        }
        if (!bfind)
        {
            return false;
        }
    }

    return true;
}
//移植
int DataCenterManager::set_devguid_type1(std::string& addr, ST_DEV_GUID stDevGuid)
{
    auto itDevguid = c_dc_addr_devguids.find(addr);
    if (itDevguid == c_dc_addr_devguids.end())
    {
        c_dc_addr_devguids.insert(std::make_pair(addr, stDevGuid));
        return 1;
    }
    else{
        c_dc_addr_devguids[addr] = stDevGuid;
        return 0;
    }

    return -1;
}

int DataCenterManager::set_devguid_type2(std::string& dev, ST_DEV_GUID stDevGuid)
{
    auto itDevguid = c_dc_guid_devguids.find(dev);
    if (itDevguid == c_dc_guid_devguids.end())
    {
        c_dc_guid_devguids.insert(make_pair(dev, stDevGuid));
        return 1;
    }
    else{
        c_dc_guid_devguids[dev] = stDevGuid;
        return 0;
    }

    return -1;
}

std::shared_ptr<ST_DEV_GUID>  DataCenterManager::get_devguid_withdev(std::string& dev)
{
    m_thdnotify_devguid1.lock();

    ST_DEV_GUID  stDevguid;
    auto itDevguid = c_dc_guid_devguids.find(dev);
    if (itDevguid != c_dc_guid_devguids.end())
    {
        stDevguid = itDevguid->second;
    }
    m_thdnotify_devguid1.unlock();

    return std::make_shared<ST_DEV_GUID>(stDevguid);
}

std::shared_ptr<ST_DEV_GUID>  DataCenterManager::get_devguid_with_devid(int64_t dev_id, bool bfrozen/* =false */)
{
    std::string addr;
    m_thdnotify_devguid2.lock();
    if(nullptr == m_cm){
        return nullptr;
    }
    auto dev = m_cm->get_terminal_device(dev_id);
    
    if (nullptr != dev)
    {
        if (dev->parent_id >0)
        {
            if (!bfrozen){            
                addr = get_comp_addr(dev->slave_addr, dev->parent_id, bfrozen);
            } else {
                addr = get_comp_addr(dev->node_id, dev->parent_id, bfrozen);
            }           
        } else {
            if (!bfrozen){
                // if(0 == strcmp(dev->channel.c_str(), "/dev/ttyRS0")){
                //     addr = "RS485-1_"+dev->slave_addr;
                // } else if(0 == strcmp(dev->channel.c_str(), "/dev/ttyRS1")){
                //     addr = "RS485-2_"+dev->slave_addr;
                // } else if(0 == strcmp(dev->channel.c_str(), "/dev/ttyRS2")){
                //     addr = "RS485-3_"+dev->slave_addr;
                // } else if(0 == strcmp(dev->channel.c_str(), "/dev/ttyRS3")){
                //     addr = "RS485-4_"+dev->slave_addr;
                // } else {
                //     addr = "HPLC_"+dev->slave_addr;
                // }
                addr = dev->slave_addr;
            }
            else{
                addr = dev->node_id;
            }            
        }
        if(0 == strcmp(dev->channel.c_str(), "/dev/ttyRS0")){
            addr = "RS485-1_"+addr;
        } else if(0 == strcmp(dev->channel.c_str(), "/dev/ttyRS1")){
            addr = "RS485-2_"+addr;
        } else if(0 == strcmp(dev->channel.c_str(), "/dev/ttyRS2")){
            addr = "RS485-3_"+addr;
        } else if(0 == strcmp(dev->channel.c_str(), "/dev/ttyRS3")){
            addr = "RS485-4_"+addr;
        } else {
            addr = "HPLC_"+addr;
        }                
    }
    m_thdnotify_devguid2.unlock();
    return get_devguid_with_devaddr(addr);
}

std::shared_ptr<ST_DEV_GUID>  DataCenterManager::get_devguid_with_devaddr(std::string& addr)
{
    ST_DEV_GUID  stDevguid;
    auto itDevguid = c_dc_addr_devguids.find(addr);
    if (itDevguid != c_dc_addr_devguids.end())
    {
        stDevguid = itDevguid->second;
    }

    return std::make_shared<ST_DEV_GUID>(stDevguid);
}

std::list<std::string> DataCenterManager::get_dc_devs()
{
    std::list<std::string> lstDevs;
    for (auto itdev = c_dc_addr_devguids.begin(); itdev != c_dc_addr_devguids.end(); itdev++)
    {
        lstDevs.push_back(itdev->second.dev);
    }

    return lstDevs;
}

std::list<ST_DEV_GUID> DataCenterManager::get_all_devguids()
{
    std::list<ST_DEV_GUID>  lstDevguid;
    for (auto itd = c_dc_addr_devguids.begin(); itd != c_dc_addr_devguids.end(); itd++)
    {
        ST_DEV_GUID stDevGuid = itd->second;
        lstDevguid.push_back(stDevGuid);
    }

    return lstDevguid;
}

void DataCenterManager::print_register_relation(std::list<ST_REGISTER_MODEL_INFO>& lstRegisterRelation)
{
    char reginfo[512]={0};
    for (auto itr = lstRegisterRelation.begin(); itr != lstRegisterRelation.end(); itr++)
    {
        for(auto itd = itr->lstRegModels.begin(); itd != itr->lstRegModels.end(); itd++){
            memset(reginfo,0,sizeof(reginfo));
            snprintf(reginfo,sizeof(reginfo),"DataCenter RegInfo:model=%d,dev=%s,addr=%d,nodeid=%s",itr->model.c_str(),itd->dev.c_str(),itd->addr.c_str(),itd->nodeID.c_str());
            CLogger::getInstance()->LOG_I("%s",reginfo);
        }
    }
       
}
int DataCenterManager::set_const_value(std::string msg){
    
    QUERY_SET_VALUE_DATAS datas;
    m_dpm.get_query_set_value_obj(datas, msg);
    
    return c_func_handle_set_value(datas);
}

void* thread_handle_const_value(void* arg){
    if(NULL == arg){
        return NULL;
    }
    DataCenterManager* dcm = (DataCenterManager*)arg;
    dcm->m_req_id++;
    NOTIFY_SET_VALUE notify;
    dcm->m_dpm.get_notify_set_value_msg(notify, dcm->c_msg);
    QUERY_SET_VALUE query;
    query.token = std::to_string(dcm->m_req_id);
    for(auto body = notify.body.begin(); body != notify.body.end(); body++){
        QUERY_SET_VALUE_BODY qb;
        qb.dev = body->dev;
        qb.totalcall = "0";
        qb.body.insert(qb.body.end(), body->body.begin(), body->body.end());
        query.body.push_back(qb);
    }
    
    std::string tmpmsg = "";
    std::string topic = DATACENTER_QUERY_SET_VALUE_PUB;
    dcm->m_dpm.get_query_set_value_msg(query, tmpmsg);
    CLogger::getInstance()->log_i("dc_query_set_value payload: %s", tmpmsg.c_str());

    dcm->m_thread_notify.lock();
    int ret = dcm->m_mosquitto_manager->publish(topic.c_str(), tmpmsg.length(), tmpmsg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        dcm->m_is_receiving = true;
        dcm->wait_recv_finished();
    }
    else {
        CLogger::getInstance()->LOG_E("datacenter dc_query_set_value [%s] error=%d.", topic.c_str(), ret);   
    }
    dcm->m_thread_notify.unlock();
}

int DataCenterManager::handle_const_value(std::string msg){
    if(msg.empty()){
        return -1;
    }
    c_msg = msg;
    pthread_t tid;
    pthread_create(&tid, NULL, thread_handle_const_value, this);
    pthread_detach(tid);    
    
    return 0;
}

void  DataCenterManager::set_handle_func(HANDLE_SET_VALUE func){
    c_func_handle_set_value = func;
}